package spendreport;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import spendreport.pojo.AccountTransation;
import spendreport.service.AccountAlertSink;
import spendreport.service.RandomNumberAccountTransactionSource;

public class FraudDetectionJob {

    public static void main(String[] args) throws Exception {
    	
    	//1.Apache Flink是一个计算框架，主要处理数据流，一般分为设置数据源，解析和操作数据，输出结果数据

    	//2.Apache Flink不关心数据来源(有边界或者无边界)，不关心数据结果输出的地点，内部只进行数据的各种处理的操作

    	//3.Source就是数据来源，Sink就是结果，通过process来执行数据处理任务，collector来收集数据
    	
    	// 设置你的执行环境，任务执行环境用于定义任务的属性、创建数据源以及最终启动任务的执行
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		 

		// 设定数据来源
		DataStream<AccountTransation> transactions = env
				.addSource(new RandomNumberAccountTransactionSource())
				.name("transactions");

		// 处理数据来源
		DataStream<AccountTransation> alerts = transactions
				.keyBy(AccountTransation::getAccountId)
				.process(new FraudDetector()) // 这里这个对象才是处理数据来源的
				.name("fraud-detector");

		// sink就是最后的结果()
		alerts
		.addSink(new AccountAlertSink())
		.name("send-alerts");

		// 最后开始执行
		env.execute("Fraud Detection");
    }
}